package org.ala.tiktools.rocketmq.comsumer;

import org.ala.tiktools.logs.LoggerFactory;
import org.ala.tiktools.rocketmq.MethodUtils;
import org.ala.tiktools.rocketmq.Namespace;
import org.ala.tiktools.rocketmq.annotations.ConsumerMethod;
import org.ala.tiktools.rocketmq.annotations.ConsumerService;
import org.ala.tiktools.rocketmq.commons.BizThreadPoolExecutor;
import org.ala.tiktools.rocketmq.config.ConfigInfo;
import org.apache.rocketmq.client.exception.MQClientException;
import org.slf4j.Logger;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.InitializingBean;
import org.springframework.beans.factory.annotation.Configurable;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.context.EnvironmentAware;
import org.springframework.context.annotation.Configuration;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Component;
import org.springframework.util.ClassUtils;
import org.springframework.util.StringUtils;

import java.lang.reflect.Method;
import java.util.Map;
import java.util.concurrent.ThreadPoolExecutor;

/**
 * Consumer初始化器
 *
 * @author ala
 * @date 2024-11-29 13:34
 */
@Configuration
public class RocketMQConsumerInitializer implements ApplicationContextAware, EnvironmentAware {


    static Logger log = LoggerFactory.getLogger("rocketmq", "init");


    /**
     *  spring环境
     */
    protected ApplicationContext context;
    /**
     *  配置
     */
    protected Environment env;
    /**
     *  执行任务的公共线程池
     */
    protected BizThreadPoolExecutor commnoThreadPool;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        this.context = applicationContext;

        initCommonThreadPool();
        initConsumers();

    }
    @Override
    public void setEnvironment(Environment environment) {
        this.env = environment;
    }


    /**
     *  初始化consumer
     */
    protected void initConsumers() {
        //  初始化公共线程池
        commnoThreadPool = new BizThreadPoolExecutor(
                "commonThreadPool",
                ConfigInfo.INSTANCE.getThreadPoolCoreSize(), ConfigInfo.INSTANCE.getThreadPoolMaxSize(),
                ConfigInfo.INSTANCE.getThreadPoolQueueSize());

        //  取所有ConsumerService注解的bean
        Map<String, Object> beans = context.getBeansWithAnnotation(ConsumerService.class);
        beans.forEach(this::initConsumer);
    }

    /**
     *  初始化公共任务线程池
     */
    protected BizThreadPoolExecutor initCommonThreadPool() {
        return new BizThreadPoolExecutor("common",
                ConfigInfo.INSTANCE.getThreadPoolCoreSize(), ConfigInfo.INSTANCE.getThreadPoolMaxSize(),
                ConfigInfo.INSTANCE.getThreadPoolQueueSize());
    }

    /**
     * 初始化Consumer
     *
     * @param beanName
     * @param bean
     */
    protected void initConsumer(String beanName, Object bean) {
        //  取注解
        ConsumerService anno = bean.getClass().getAnnotation(ConsumerService.class);

        //  创建bean独享线程池
        BizThreadPoolExecutor beanThreadPool;
        if (anno.coreSize() > 0) {
            int coreSize = anno.coreSize(),
                maxSize = anno.maxSize() <= 0 ? commnoThreadPool.getCoreSize() : anno.maxSize(),
                queueSize = anno.queueSize() <= 0 ? commnoThreadPool.getQueueSize() : anno.queueSize();
            beanThreadPool = new BizThreadPoolExecutor(beanName, coreSize, maxSize, queueSize);
        } else {
            beanThreadPool = commnoThreadPool;
        }

        //  namesrc地址
        String namesrvAddrs;
        if (!anno.namesrvAddrs().equals(Namespace.NULL)) {
            namesrvAddrs = ConfigInfo.INSTANCE.replaceEnv(anno.namesrvAddrs());
        } else {
            namesrvAddrs = ConfigInfo.INSTANCE.getNamesrvAddr();
        }
        //  group
        String group = ConfigInfo.INSTANCE.replaceEnv(anno.group());
        //  topic
        String topic = ConfigInfo.INSTANCE.replaceEnv(anno.topic());

        //  解析bean的所有含ConsumerMethod注解的方法
        Map<ConsumerMethod, Method> methods = MethodUtils.findMethods(bean.getClass(), ConsumerMethod.class);

        //  每个方法对应一个执行器，每个执行器会初始化一个Consumer
        methods.forEach((a, method) -> {
            try {
                initConsumerExecutor(beanName, bean, method,
                                                beanThreadPool,
                                                namesrvAddrs, group, topic, a);
            } catch (MQClientException e) {
                log.error("[RocketMQConsumerInitializer initConsumer] 初始化Consumer异常 " + beanName + " namesrvAddrs:" + namesrvAddrs, e);
            }
        });
    }
    protected void initConsumerExecutor(String beanName, Object bean, Method method,
                                        BizThreadPoolExecutor beanThreadPool,
                                        String namesrcAddrs, String group, String topic,
                                        ConsumerMethod anno) throws MQClientException {
        //  group, topic, tag
        if (!anno.group().equals(Namespace.NULL)) {
            group = ConfigInfo.INSTANCE.replaceEnv(anno.group());
        }
        if (!anno.topic().equals(Namespace.NULL)) {
            topic = ConfigInfo.INSTANCE.replaceEnv(anno.topic());
        }
        String tag = "*";
        if (!StringUtils.isEmpty(tag)) {
            tag = ConfigInfo.INSTANCE.replaceEnv(anno.tag());
        }
        //  broadCasting, orderly
        boolean broadCasting = anno.broadCasting();
        boolean orderly = anno.orderly();


        //  创建executor独享线程池
        BizThreadPoolExecutor executorThreadPool = beanThreadPool;
        if (anno.coreSize() > 0) {
            int coreSize = anno.coreSize(),
                maxSize = anno.maxSize() <= 0 ? executorThreadPool.getCoreSize() : anno.maxSize(),
                queueSize = anno.queueSize() <= 0 ? executorThreadPool.getQueueSize() : anno.queueSize();
            executorThreadPool = new BizThreadPoolExecutor(beanName + "-" + method.getName(), coreSize, maxSize, queueSize);
        }

        //  初始化consumer执行器
        ConsumerExecutor.build(beanName, bean, method, executorThreadPool,
                namesrcAddrs, group, topic, tag, broadCasting, orderly);
    }
}
